package edu.vt.ece.rt.quorum.messsages.types;

import java.util.HashSet;
import java.util.Set;

import org.jgroups.Address;
import org.jgroups.Message;

import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.Cluster;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.QMessage;
import edu.vt.ece.rt.quorum.messsages.QMessage.MessageType;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Read Request to all the nodes in read Quorum
 * @author Sudhanshu
 */

public class ReadRequest extends QMessage implements Runnable {
	String Id;			//Specify the Object Id to be received
	public Context txn; 		//Specify the transaction number, Can be Converted to Context Class, if some MetaData required
	Long txnIdParent; 		//parent txn ID
	boolean read;		//Specify read or write 
	int requesterID;		//Specify the cluster on which response will be unicast back


	public ReadRequest(String id, Context t, Long tidParent, boolean rw, int rid, String mId) {
		super(MessageType.ReadRequest);
		Id = id;
		txn = t;
		txnIdParent = tidParent;
		read = rw;
		requesterID = rid;
		messageId = mId;
	}

	@Override
	public void run() {
		//Get Object
		Long txnId = txn.getContextId();
		Logger.debug("Running the "+ (read?"Read":"write")+" Request of "+ Id + " for " + txnId+ " from "+requesterID +" with parent as "+txnIdParent);
		QObject qObj = null;
		boolean abort = false;
		
		synchronized (Resource.waitObj) {
			Context c = ContextDelegator.getContext(txnId);
			c.setParent(txnIdParent);
			c.setPriority(txn.getPriority());
			
//			if (Id.contains("grid")) {
//				createGrid(txnId);
//				return;
//			}
			
			qObj = Resource.getQObject(Id);
			Long abortTxn = validation (txn);
			boolean inValid = (abortTxn != null);
			boolean objectProtected = qObj.isProtected(txnId);

			if (inValid || objectProtected) {
				// in any case remove the current transaction from lists
				Resource.rmFrLists(ContextDelegator.getContext(txnId));
				ContextDelegator.getContext(txnId).setAborted();

				abort = true;
				qObj = qObj.clone();

				if (inValid) {
					qObj.setAbortTxn(abortTxn);
					// remove the parent transaction
					
					if (txn.isChild() && (abortTxn.longValue() == txnIdParent.longValue())) {
						Resource.rmFrLists(ContextDelegator.getContext(abortTxn));
						ContextDelegator.getContext(abortTxn).setAborted();
					}
				} else if (objectProtected) {
					qObj.setAbortTxn(txnId);
				}
			} else {
				qObj = qObj.clone();
				qObj.setCloneOwner(txnId);
			}

			//Add Context to potential read-write list
			if (!abort) {
				if(read == true){
					if (c.isRoot()) {
						Resource.addPR(Id, c);
						c.addToReadSet(qObj);
					}
					
					Logger.debug("Added "+txnId+ " as potential reader of "+qObj.getId());
				} else{
					if (c.isRoot()) {
						Resource.addPW(Id, c);
						c.addToWriteSet(qObj);
					}
					Logger.debug("Added "+txnId+ " as potential writer of "+qObj.getId());
				}
			}

		}
		//Send back the response to node
		try {
			//			if (qObj == null) {
			//				Logger.debug("Repling with null object");
			//			}
			//			if(read)
			ClustersManager.getCluster(requesterID+Names.readQuorum).send(new Message(getSrcAddr(), new ReadResponse(Id, txnId, qObj, read)));
			//			else
			//				ClustersManager.getCluster(requesterID+Names.writeQuorum).send(new Message(getSrcAddr(), new ReadResponse(Id, txnId, qObj, read)));
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public void createGrid(Long txnId) {
		
		
//		Grid grid = new Grid(Long.toString(txnId));
//		
//		for (int x =0 ; x < gridX; x++) {
//			for (int y = 0; y < gridY; y++) {
//				for (int z = 0; z < gridZ; z++) {
//					QObject o = Resource.getQObject(x+"-"+y+"-"+z);
//					Cell c = (Cell)o.clone();
//					grid.add(c);
//				}
//			}
//		}

		try {
//			ClustersManager.getCluster(requesterID+Names.readQuorum).send(new Message(getSrcAddr(), 
//					new ReadResponse(Id, txnId, grid, read)));
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public Long validation(Context remoteContext){

		Set<Context> conflictSet = new HashSet<Context>();
		boolean commit = true;
		Long abortTxn = Long.MAX_VALUE;
		Long txn = remoteContext.getContextId();

		Set<QObject> dataSet = remoteContext.getDataSet();

		Logger.debug("size of data "+dataSet.size());
		for (QObject obj : dataSet) {
			// Check local object status and version comparison
			Logger.debug("txn : "+txn+" Checking for obj "+ obj.getId() +" clone owner "+obj.getCloneOwner());
			
			if (Resource.getQObject(obj.getId()).isProtected(txn)
					|| Resource.isStaleVersion(obj)) {
				Logger.debug("txn : "+txn+" Invalid for obj "+ obj.getId());
				// True, means abort this transaction
				ContextDelegator.getContext(txn).setAborted();
				commit = false;

				if (obj.getCloneOwner() < abortTxn) { 
					abortTxn = obj.getCloneOwner();
					conflictSet.clear();
					conflictSet.add(ContextDelegator.getContext(abortTxn));
					Logger.debug("txn : "+txn+" updating new clone owner to parent "+abortTxn);
					Resource.rmFrLists(ContextDelegator.getContext(abortTxn));
				}

			}
		}

		if (commit) {
			return null;
		} else {
			Logger.debug("aborting "+abortTxn+" as part of read validation");
			return abortTxn;
		}

	}

}
